package com.fafi.rxjava2.observeon.nobackpress;

import com.fafi.rxjava1.subscribeon.Switcher;
import com.fafi.rxjava2.Callee;
import com.fafi.rxjava2.Release;
import com.fafi.rxjava2.operator.Caller;
import com.fafi.rxjava2.operator.CallerWithUpstream;

import java.util.LinkedList;
import java.util.Queue;

public class CallerCallbackOn<T> extends CallerWithUpstream<T, T> {

    private final Switcher mSwitcher;

    public CallerCallbackOn(Caller<T> source, Switcher mSwitcher) {
        super(source);
        this.mSwitcher = mSwitcher;
    }

    @Override
    protected void callActual(Callee<T> callee) {
        source.call(new CallbackOnCallee<T>(callee, mSwitcher));
    }

    private static final class CallbackOnCallee<T> implements Callee<T>, Runnable {
        private final Callee<T> mCallee;
        private final Switcher.Worker worker;
        private final Queue<T> tQueue = new LinkedList<>();

        public CallbackOnCallee(Callee<T> mCallee, Switcher switcher) {
            this.mCallee = mCallee;
            this.worker = switcher.createWorker();
        }

        @Override
        public void onCall(Release release) {

        }

        @Override
        public void onReceive(T t) {
            tQueue.offer(t);
            switches();
        }


        @Override
        public void onCompleted() {

        }

        @Override
        public void onError(Throwable t) {

        }

        @Override
        public void run() {
            T t = tQueue.poll();
            mCallee.onReceive(t);
        }

        private void switches() {
            worker.switches(this);
        }
    }
}